//
// Created by twetec on 25-7-25.
//

#include "perception_rpc_caller_sustain_probe_info.h"

#include <grpcpp/grpcpp.h>

#include <libcron/Cron.h>

#include <network_prober/sys_config/sys_config.h>

using namespace std::chrono_literals;

namespace network_prober::rpc
{
/**
 * 拦截器，用于感知主机验证
 */
class SustainProbeInfoPerceptionHostAuthInterceptor : public grpc::experimental::Interceptor
{
public:
    SustainProbeInfoPerceptionHostAuthInterceptor(grpc::experimental::ClientRpcInfo* info) : m_clientRPCInfoPtr(info)
    {}

    /**
     * 拦截器拦截的方法，它会在很多时机(hook)下拦截，所以叫methods
     * @param methods 多个时机方法
     */
    void Intercept(grpc::experimental::InterceptorBatchMethods* methods) override
    {
        // 客户端接收到消息之后
        // if (methods->QueryInterceptionHookPoint(grpc::experimental::InterceptionHookPoints::POST_RECV_MESSAGE)) {
        //     m_clientRPCInfoPtr->method();
        // }
        // std::cout << "调用了拦截器" << std::endl;
        methods->Proceed();
    }

private:
    grpc::experimental::ClientRpcInfo* m_clientRPCInfoPtr;
};

/**
 * 拦截器，工厂方法
 */
class SustainProbeInfoPerceptionHostAuthInterceptorFactory : public grpc::experimental::ClientInterceptorFactoryInterface
{
public:
    grpc::experimental::Interceptor* CreateClientInterceptor(grpc::experimental::ClientRpcInfo* info) override
    {
        return new SustainProbeInfoPerceptionHostAuthInterceptor(info);
    }
};

SustainProbeInfoRPCClientReactor::SustainProbeInfoRPCClientReactor(network_perception::rpc::NetworkPerceptionServer::Stub *RPCClientStubPtr)
{
    // 将clientContext和自己传入RPC的方法(设置代理方法)
    RPCClientStubPtr->async()->sustainProbeInfo(std::addressof(m_clientContext), std::addressof(probeInfoResponse), this);
}

void SustainProbeInfoRPCClientReactor::startCalling()
{
    // Use a hold since some StartWrites are invoked indirectly from a
    // delayed lambda in OnWriteDone rather than directly from the reaction
    // itself
    // this->AddHold();
    // 在RPC启动时，没有要发送的内容，需要调用startWritesDone()，通知服务器已写入完成
    // this->StartWritesDone();
    // 建立并激活指令连接rpc
    this->StartCall();
}

grpc::Status SustainProbeInfoRPCClientReactor::awaitCallingDone()
{
    std::unique_lock<std::mutex> lock(m_mutex);
    m_condition.wait(lock, [this](){ return callingDone; });
    return std::move(m_callingStatus);
}

/**
 * 当一次写入完成时调用
 * @param ok 写入是否成功执行
 */
void SustainProbeInfoRPCClientReactor::OnWriteDone(bool ok)
{
    SPDLOG_INFO("===== SustainProbeInfo OnWriteDone: {} =====", ok);
}

/**
 * 响应RPC完成。我们将在 OnDone() 中执行最终清理
 */
void SustainProbeInfoRPCClientReactor::OnDone(const grpc::Status& status)
{
    SPDLOG_INFO("===== SustainProbeInfo caller结束(onDone): {} =====", status.ok());
    // 如果没正常结束，可以拿到错误信息
    // status.error_message()
    std::unique_lock<std::mutex> lock(m_mutex);
    callingDone = true;
    m_callingStatus = status;
    m_condition.notify_one();
}

PerceptionRPCCallerSustainProbeInfo::PerceptionRPCCallerSustainProbeInfo()
{
    // 1、获取感知系统配置信息
    std::string perceptionIP = sys_config::SysConfig::getInstance().perceptionIP();
    std::size_t perceptionPort = sys_config::SysConfig::getInstance().perceptionPort();

    // 2、构建通道参数
    grpc::ChannelArguments channelArguments;
    // 连接感知
    channelArguments.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, 20 * 1000 /*45 sec*/);  // 每隔多长时间发送一次PING
    channelArguments.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 10 * 1000 /*10 sec*/); // 发送PING后，响应超时时间 [如果发送方在此时间内未收到确认，它将关闭连接]
    channelArguments.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);    // 是否允许发送心跳数据
    // 断连重接
    channelArguments.SetInt(GRPC_ARG_ENABLE_RETRIES, 1);    // 允许重连
    channelArguments.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, 100); // 初次重连间隔时间
    channelArguments.SetInt(GRPC_ARG_MIN_RECONNECT_BACKOFF_MS, 100);    // 最小重连间隔时间
    channelArguments.SetInt(GRPC_ARG_MAX_RECONNECT_BACKOFF_MS, 100);    // 最大重连间隔时间
    // HTTP2设置
    channelArguments.SetInt(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA, 0);
    channelArguments.SetInt(GRPC_ARG_HTTP2_MIN_SENT_PING_INTERVAL_WITHOUT_DATA_MS, 30 * 1000);
    channelArguments.SetInt(GRPC_ARG_HTTP2_BDP_PROBE, 1);
    channelArguments.SetInt(GRPC_ARG_HTTP2_MAX_FRAME_SIZE, 16 * 1024 * 1024); // 16M
    // 内存管理
    channelArguments.SetInt(GRPC_ARG_MAX_SEND_MESSAGE_LENGTH, -1); // 通道可发送的消息最大字节数，-1表示没限制
    channelArguments.SetInt(GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH, -1);   // 通道可接受的消息最大字节数，-1表示没限制

    // 3、定义拦截器
    std::vector<std::unique_ptr<grpc::experimental::ClientInterceptorFactoryInterface>> interceptorCreators;
    // 添加拦截器
    interceptorCreators.push_back(std::make_unique<SustainProbeInfoPerceptionHostAuthInterceptorFactory>());

    // 4、用通道参数创建通道
    std::shared_ptr<grpc::Channel> channel = grpc::experimental::CreateCustomChannelWithInterceptors(
        perceptionIP + ":" + std::to_string(perceptionPort),
        grpc::InsecureChannelCredentials(),
        channelArguments,
        std::move(interceptorCreators)
    );

    // 4、创建远程调用代理(存根对象)
    m_RPCClientStubPtr = network_perception::rpc::NetworkPerceptionServer::NewStub(channel);
    // 构建reactor
    m_RPCClientReactorPtr = std::make_unique<SustainProbeInfoRPCClientReactor>(m_RPCClientStubPtr.get());
}

void PerceptionRPCCallerSustainProbeInfo::_sustainProbeInfo(std::stop_token stopToken)
{
    // 定时任务控制对象
    libcron::Cron cron;

    // 给定时任务添加一个工作计划 [58 * * * * ?] -> 从右往左读 -> 无效周几 每年 每月 每日 每时 每分 的第58秒 执行
    // 等2个0s周期后，观测任务就开始执行
    cron.add_schedule("SUSTAIN_PROBE_INFO", "*/4 * * * * ?", [this, &cron](const libcron::TaskInformation& taskInfo) {
        // 清空请求
        m_RPCClientReactorPtr->probeInfoResult.Clear();

        // 设置请求
        // 密钥
        m_RPCClientReactorPtr->probeInfoResult.set_secretkey(sys_config::SysConfig::getInstance().proberSecretKey());

        // 设置CPU使用百分比
        m_RPCClientReactorPtr->probeInfoResult.set_cpuusage(sys_config::SysConfig::getInstance().proberCPUUsage());
        // 设置RAM使用百分比
        m_RPCClientReactorPtr->probeInfoResult.set_ramusage(sys_config::SysConfig::getInstance().proberMemoryUsage());
        // 设置DISK使用百分比
        m_RPCClientReactorPtr->probeInfoResult.set_diskusage(sys_config::SysConfig::getInstance().proberDiskUsage());
        // 设置DISK空间
        m_RPCClientReactorPtr->probeInfoResult.set_disksize(sys_config::SysConfig::getInstance().proberDiskSpaceGB());
        // 设置当前系统时间
        m_RPCClientReactorPtr->probeInfoResult.set_time(std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch()).count());

        // 发送请求
        m_RPCClientReactorPtr->StartWrite(std::addressof(m_RPCClientReactorPtr->probeInfoResult));
    });

    // 作为定时任务的时钟，进行定时
    while(!stopToken.stop_requested() && !m_RPCClientReactorPtr->callingDone) {
        cron.tick(); // 保证至少每秒调用一次，避免错过任何调度
        std::this_thread::sleep_for(500ms); // 等待500ms,再调度一次工作计划
    }

    // 移除定时任务的所有工作计划
    cron.clear_schedules();
}

void PerceptionRPCCallerSustainProbeInfo::callImpl()
{
    // instruction双向流通道的状态
    grpc::Status grpcChannelStatus;
    do {
        // 建立流连接
        m_RPCClientReactorPtr->startCalling();

        // 持续获取系统运行参数，然后上报
        // 创建并发起取消请求的源
        std::stop_source stopSource;
        std::stop_token stopToken = stopSource.get_token();
        std::jthread sustainProbeInfoThread{&PerceptionRPCCallerSustainProbeInfo::_sustainProbeInfo, this, stopToken};

        // 等待调用结束（监听变量）
        grpcChannelStatus = m_RPCClientReactorPtr->awaitCallingDone();

        // 当channel结束时，请求停止子线程运行
        stopSource.request_stop();

        // 如果正常结束则停止
        if (grpcChannelStatus.ok()) break;

        // 如果没有正常结束，则重新发起连接
        // 日志输出
        SPDLOG_INFO("sustain_probe_info 客户端流未正常结束: {} - {}，重新发起连接..", static_cast<std::int8_t>(grpcChannelStatus.error_code()), grpcChannelStatus.error_message());

        // 1、获取感知系统配置信息
        std::string perceptionIP = sys_config::SysConfig::getInstance().perceptionIP();
        std::size_t perceptionPort = sys_config::SysConfig::getInstance().perceptionPort();
        // 2、构建通道参数
        grpc::ChannelArguments channelArguments;
        // 连接感知
        channelArguments.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, 20 * 1000 /*45 sec*/);  // 每隔多长时间发送一次PING
        channelArguments.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 10 * 1000 /*10 sec*/); // 发送PING后，响应超时时间 [如果发送方在此时间内未收到确认，它将关闭连接]
        channelArguments.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);    // 是否允许发送心跳数据
        // 断连重接
        channelArguments.SetInt(GRPC_ARG_ENABLE_RETRIES, 1);    // 允许重连
        channelArguments.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, 100); // 初次重连间隔时间
        channelArguments.SetInt(GRPC_ARG_MIN_RECONNECT_BACKOFF_MS, 100);    // 最小重连间隔时间
        channelArguments.SetInt(GRPC_ARG_MAX_RECONNECT_BACKOFF_MS, 100);    // 最大重连间隔时间
        // HTTP2设置
        channelArguments.SetInt(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA, 0);
        channelArguments.SetInt(GRPC_ARG_HTTP2_MIN_SENT_PING_INTERVAL_WITHOUT_DATA_MS, 30 * 1000);
        channelArguments.SetInt(GRPC_ARG_HTTP2_BDP_PROBE, 1);
        channelArguments.SetInt(GRPC_ARG_HTTP2_MAX_FRAME_SIZE, 16 * 1024 * 1024); // 16M
        // 内存管理
        channelArguments.SetInt(GRPC_ARG_MAX_SEND_MESSAGE_LENGTH, -1); // 通道可发送的消息最大字节数，-1表示没限制
        channelArguments.SetInt(GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH, -1);   // 通道可接受的消息最大字节数，-1表示没限制
        // 3、定义拦截器
        std::vector<std::unique_ptr<grpc::experimental::ClientInterceptorFactoryInterface>> interceptorCreators;
        // 添加拦截器
        interceptorCreators.push_back(std::make_unique<SustainProbeInfoPerceptionHostAuthInterceptorFactory>());
        // 4、用通道参数创建通道
        std::shared_ptr<grpc::Channel> channel = grpc::experimental::CreateCustomChannelWithInterceptors(
            perceptionIP + ":" + std::to_string(perceptionPort),
            grpc::InsecureChannelCredentials(),
            channelArguments,
            std::move(interceptorCreators)
        );
        // 5、创建远程调用代理(存根对象)
        m_RPCClientStubPtr = network_perception::rpc::NetworkPerceptionServer::NewStub(channel);
        // 构建reactor
        m_RPCClientReactorPtr = std::make_unique<SustainProbeInfoRPCClientReactor>(m_RPCClientStubPtr.get());

    } while (!grpcChannelStatus.ok());
}
}